package org.jeecg.modules.service.impl;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.jeecg.modules.service.MessageListener;
import org.jeecg.modules.service.MqttService;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 *MQTT连接服务基本实现
 * @author tanxy
 * @date 2022年11月27日 21:59
 */
@Slf4j
public class MqttServiceImpl implements MqttService, MqttCallbackExtended {

    private String url;
    private String clientId;
    private String userName;
    private String password;

    private MessageListener listener;

    private boolean connected = false;

    private Integer qos = 1;

    private MqttClient client;

    private ExecutorService executor = null;


    /**
     * 构造函数
     *
     * @param url      MQTT地址
     * @param clientId 客户端标识
     * @param userName 用户名
     * @param password 密码
     */
    public MqttServiceImpl(String url, String clientId, String userName, String password) {
        this.url = url;
        this.clientId = clientId;
        this.userName = userName;
        this.password = password;

        executor = Executors.newFixedThreadPool(5);
    }

    public void setListener(MessageListener listener) {
        this.listener = listener;
    }

    public void start() throws Exception {
        if (client != null) {
            client.close();
        }
        client = new MqttClient(url, clientId, new MemoryPersistence());

        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(true);
        options.setAutomaticReconnect(true);
        options.setConnectionTimeout(30);
        options.setKeepAliveInterval(120);
        if (!StringUtils.isBlank(userName)) {
            options.setUserName(userName);
        }
        if (!StringUtils.isBlank(password)) {
            options.setPassword(password.toCharArray());
        }

        client.setCallback(this);
        client.connect(options);
    }

    public boolean isConnected() {
        return connected;
    }

    public void publish(String topic, String msg) throws Exception {
        publish(topic, msg.getBytes());
    }

    public void publish(final String topic, final byte[] msg) throws Exception {
        executor.execute(new Runnable() {

            public void run() {
                if (!connected) {
                    return;
                }
                MqttMessage message = new MqttMessage(msg);
                message.setQos(qos);

                try {
                    client.publish(topic, message);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

    }

    public void stop() {
        if (client != null) {
            try {
                client.disconnect();
                client.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        client = null;
    }

    public void connectionLost(Throwable cause) {
        System.out.println("mqtt connection lost!");
        log.error(cause.getLocalizedMessage());
        cause.printStackTrace();

        if(client.isConnected()){
            return;
        }

        connected = false;
        if (listener != null) {
            listener.handleConnectStatus(connected);
        }
    }

    public void messageArrived(String topic, MqttMessage message) throws Exception {
        if (listener != null) {
            listener.handleMessage(topic, message.getPayload());
        }
    }

    public void deliveryComplete(IMqttDeliveryToken token) {
        System.out.println("deliveryComplete---------" + token.isComplete());
    }

    public void connectComplete(boolean reconnect, String serverURI) {
        connected = true;
        if (listener != null) {
            listener.handleConnectStatus(connected);
        }

        try {
            client.subscribe(TopicUtil.syncTimeReqTopic, qos);
            client.subscribe(TopicUtil.heartbeatTopic, qos);
            client.subscribe(TopicUtil.gameRealTimeDataTopic, qos);
            client.subscribe(TopicUtil.calling, qos);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
